package nsqproducer

import (
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/nsqio/go-nsq"
	"io/ioutil"
	"math/big"
	"net/http"
	"runtime"
	"strconv"
	"sync"
	"time"
)

func init() {
	fmt.Println(welcome)
}

func (pushNode *PushNode) InitNsqProducer(nsqlookupdAddr string) {
	pushNode.setDefaultArg()
	pushNode.newNsqProducer(nsqlookupdAddr)
	pushNode.checkNsqdStatus(nsqlookupdAddr)
}

func (pushNode *PushNode) setDefaultArg() {

	pushNode.ConnNodeDepth = make(chan struct{}, runtime.NumCPU()*10)

	if pushNode.NsqConf == nil {
		pushNode.NsqConf = nsq.NewConfig()
	}

	pushNode.nextNodeIndex = -1
	pushNode.nsqdFristProducerNum = -1
	pushNode.nodeIndexLock = &sync.Mutex{}

	if pushNode.RetryNodeDepth == 0 {
		pushNode.RetryNodeDepth = 2
	}

	if pushNode.NsqdProducerMaxNum == 0 {
		pushNode.NsqdProducerMaxNum = 3
	}

	if pushNode.CheckInterval == 0 {
		pushNode.CheckInterval = 10 * time.Second
	}

	if pushNode.RetryPushInterval == 0 {
		pushNode.RetryPushInterval = 2 * time.Second
	}

	if pushNode.RetryPushMaxNum == 0 {
		pushNode.RetryPushMaxNum = 3
	}
}

func (pushNode *PushNode) newNsqProducer(nsqlookupdAddr string) {
	pushNode.getNsqdList(nsqlookupdAddr)
	pushNode.parallelConnToNsqd()
	pushNode.nodesTemp = NsqdNodes{}
}

func (pushNode *PushNode) parallelConnToNsqd() {
	pushNode.nodeList = sync.Map{}

	for nodeIndex, node := range pushNode.nodes.Producers {
		pushNode.ConnNodeDepth <- struct{}{}
		nsqdAddr := node.BroadcastAddress + ":" + strconv.Itoa(node.TcpPort)
		go pushNode.connToNsqd(nodeIndex, nsqdAddr)
	}
}

func (pushNode *PushNode) connToNsqd(nodeIndex int, nsqdAddr string) {

	if pushNode.nodes.Producers[nodeIndex].nsqdStatus {
		pushNode.setNsqdStatus(nodeIndex, true)
		fmt.Println(time.Now().Format("2006-01-02 15:04:05"), " INF ", nsqdAddr, " connection already exists")
		return
	}

	pushNode.nodes.Producers[nodeIndex].producer, _ = nsq.NewProducer(nsqdAddr, pushNode.NsqConf)
	if pushNode.nodes.Producers[nodeIndex].producer.Ping() != nil {

		pushNode.nodes.Producers[nodeIndex].nsqdAddr = nsqdAddr
		pushNode.setNsqdStatus(nodeIndex, false)
		fmt.Println(time.Now().Format("2006-01-02 15:04:05"), " ERR ", nsqdAddr, " connection failed")
		return
	}
	pushNode.nodes.Producers[nodeIndex].nsqdAddr = nsqdAddr
	pushNode.setNsqdStatus(nodeIndex, true)

	fmt.Println(time.Now().Format("2006-01-02 15:04:05"), " INF ", nsqdAddr, " connection complete")

	<-pushNode.ConnNodeDepth
}

func (pushNode *PushNode) getNsqdList(lookupAddr string) {
	var (
		client = &http.Client{Timeout: 10 * time.Second}
		qyurl  = "http://" + lookupAddr + "/nodes"
	)

	response, err := client.Get(qyurl)
	if err != nil {
		panic(err)
	}

	body, err := ioutil.ReadAll(response.Body)
	if err != nil {
		panic(err)
	}

	if err = json.Unmarshal(body, &pushNode.nodesTemp); err != nil {
		panic(err)
	}

	pushNode.nsqdCountNum = len(pushNode.nodesTemp.Producers)

	pushNode.getTopicNsqdList()
}

func (pushNode *PushNode) getTopicNsqdList() {

	if pushNode.nsqdCountNum <= pushNode.NsqdProducerMaxNum {

		if pushNode.nsqdFristProducerNum == -1 {
			pushNode.nodes.Producers = pushNode.nodesTemp.Producers
			pushNode.nsqdNowProducerNum = pushNode.nsqdCountNum
			pushNode.nsqdFristProducerNum = pushNode.nsqdNowProducerNum
			return
		}

		pushNode.nsqdTempProducerNum = pushNode.nsqdCountNum
		pushNode.getOptimalNsqd()

		return
	}
	pushNode.nsqdTempProducerNum = pushNode.NsqdProducerMaxNum
	pushNode.getOptimalNsqd()
}

func (pushNode *PushNode) getOptimalNsqd() {
	var topicNodeList []NsqdNode
	useNodeList := make(map[int]int)

	for nodeIndex, node := range pushNode.nodesTemp.Producers {

		if len(topicNodeList) < pushNode.nsqdTempProducerNum {

			nodeAddr := node.BroadcastAddress + ":" + strconv.Itoa(node.TcpPort)
			if indexStruct, ok := pushNode.nodeList.Load(nodeAddr); ok && indexStruct.(NodeIndex).Status {
				topicNodeList = append(topicNodeList, pushNode.nodes.Producers[indexStruct.(NodeIndex).Index])
				useNodeList[nodeIndex] = 1
			}

		} else {
			break
		}

	}

	for nodeIndex, node := range pushNode.nodesTemp.Producers {
		if len(topicNodeList) < pushNode.nsqdTempProducerNum {
			if _, ok := useNodeList[nodeIndex]; !ok {

				for _, topicName := range node.Topics {
					if pushNode.Topic == "" || topicName == pushNode.Topic {
						topicNodeList = append(topicNodeList, node)
						useNodeList[nodeIndex] = 1
						break
					}
				}

			}
		} else {
			break
		}
	}

	if len(topicNodeList) < pushNode.nsqdTempProducerNum {
		topicNodeList = append(
			topicNodeList,
			pushNode.getRandNode(useNodeList)...,
		)
	}

	pushNode.nodes.Producers = topicNodeList
	pushNode.nsqdNowProducerNum = pushNode.nsqdTempProducerNum
	if pushNode.nsqdFristProducerNum == -1 {
		pushNode.nsqdFristProducerNum = pushNode.nsqdTempProducerNum
	}
}

func (pushNode *PushNode) getRandNode(useNodeList map[int]int) (randNodeList []NsqdNode) {
	n := 0
	randMixNum := len(pushNode.nodesTemp.Producers)
	getRandNum := pushNode.nsqdTempProducerNum - len(useNodeList)

	for {
	start:
		randNum64, _ := rand.Int(rand.Reader, big.NewInt(int64(randMixNum)))
		randIndex, err := strconv.Atoi(randNum64.String())
		if err != nil {
			goto start
		}

		if useNodeList[randIndex] == 1 {
			goto start
		}

		randNodeList = append(randNodeList, pushNode.nodesTemp.Producers[randIndex])
		useNodeList[randIndex] = 1
		n++
		if n == getRandNum {
			break
		}
	}
	return randNodeList
}

func (pushNode *PushNode) nextPushNodeIndex() (int, error) {
	pushNode.nodeIndexLock.Lock()
	oldNodeIndex := pushNode.nextNodeIndex
start:

	if pushNode.nextNodeIndex >= pushNode.nsqdNowProducerNum-1 {
		pushNode.nextNodeIndex = 0
	} else {
		pushNode.nextNodeIndex++
	}

	if !pushNode.nodes.Producers[pushNode.nextNodeIndex].nsqdStatus {
		if oldNodeIndex == pushNode.nextNodeIndex {
			return -1, errors.New("no nsqd available")
		}
		goto start
	}
	pushNode.nodeIndexLock.Unlock()
	return pushNode.nextNodeIndex, nil
}

func (pushNode *PushNode) PushMessage(topic string, message []byte) error {
	retryGetDepth := 0
retryGetIndex:
	nodeIndex, err := pushNode.nextPushNodeIndex()
	if err != nil {
		return err
	}
	retryPushNum := 0
retryPush:
	if !pushNode.nodes.Producers[nodeIndex].nsqdStatus {
		goto retryGetIndex
	}

	if pushErr := pushNode.nodes.Producers[nodeIndex].producer.Publish(topic, message); err != nil {
		time.Sleep(pushNode.RetryPushInterval)
		retryPushNum++

		if retryPushNum == pushNode.RetryPushMaxNum {
			retryGetDepth++
			pushNode.setNsqdStatus(nodeIndex, false)
			if retryGetDepth == pushNode.RetryNodeDepth {
				return pushErr
			}
			goto retryGetIndex
		}
		goto retryPush
	}
	return nil
}

func (pushNode *PushNode) checkNsqdStatus(nsqlookupdAddr string) {
	go func() {
		for {
			select {
			case <-time.NewTimer(pushNode.CheckInterval).C:

				if pushNode.nsqdFristProducerNum != pushNode.nsqdNowProducerNum {
					pushNode.newNsqProducer(nsqlookupdAddr)
					break
				}

				for index, nsqdNode := range pushNode.nodes.Producers {

					if nsqdNode.producer.Ping() != nil {
						if n := pushNode.retryNewProducer(&nsqdNode, index, pushNode.NsqConf); n == 0 {
							pushNode.newNsqProducer(nsqlookupdAddr)
							break
						}
					}
				}

			}
		}
	}()
}

func (pushNode *PushNode) retryNewProducer(nsqdNode *NsqdNode, index int, nsqConfig *nsq.Config) int {
	retry := 0
retryStart:
	nsqdNode.producer, _ = nsq.NewProducer(nsqdNode.nsqdAddr, nsqConfig)

	if nsqdNode.producer.Ping() != nil {

		retry++
		if retry == pushNode.RetryPushMaxNum {
			pushNode.setNsqdStatus(index, false)

			return 0
		}
		time.Sleep(pushNode.RetryPushInterval)
		goto retryStart

	}
	pushNode.setNsqdStatus(index, true)
	return 1
}

func (pushNode *PushNode) setNsqdStatus(index int, status bool) {
	pushNode.nodes.Producers[index].nsqdStatus = status
	nodeIndex := NodeIndex{
		Index:  index,
		Status: status,
	}
	pushNode.nodeList.Store(pushNode.nodes.Producers[index].nsqdAddr, nodeIndex)
}
